package com.link2room.bumblebee.facade.impl;

import com.link2room.bumblebee.facade.ImnsLogFacadeService;
import com.link2room.common.constant.DefaultValue;
import com.link2room.common.service.ICommonService;
import com.link2room.bumblebee.dto.param.MnsConsumerLogParamDto;
import com.link2room.bumblebee.dto.param.MnsProducterLogParamDto;
import com.link2room.bumblebee.entity.MnsConsumLogCountEntity;
import com.link2room.bumblebee.entity.MnsProducteLogCountEntity;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.redis.util.RedisLockRegistry;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * Created by LN on 2017/9/12.10:44
 */
@Component("mnsLogFacadeService")
public class MnsLogFacadeServiceImpl implements ImnsLogFacadeService {

    private static ThreadLocal<Map<String, Long>> threadCounter = new ThreadLocal<>();
    private static final Object lockObj = new Object();

    @Resource
    private ICommonService commonService;

    @Autowired
    private RedisLockRegistry mnsLogLock;

    @Override
    public void logProducter(MnsProducterLogParamDto paramDto) throws Exception {
        String producter = paramDto.getProducteMnsQueue().getProducter();
        String queueName = paramDto.getProducteMnsQueue().getQueueName();
        String key = "P" + producter + queueName;
        counterAdd(key);

        this.concurrentlog(key,
                () -> {
                    Long count = threadCounter.get().get(key);
                    MnsProducteLogCountEntity find = new MnsProducteLogCountEntity();
                    find.setProducter(producter);
                    find.setQueueName(queueName);
                    MnsProducteLogCountEntity qres = commonService.findOneEQ(MnsProducteLogCountEntity.class, find);
                    if (qres == null) {
                        qres = new MnsProducteLogCountEntity();
                        qres.setLhotelGroupCode(DefaultValue.LHOTEL_GROUP_CODE);
                        qres.setQueueName(queueName);
                        qres.setProducter(producter);
                        qres.setCount(0L);
                    }
                    long nCount = qres.getCount() + count;
                    qres.setCount(nCount);
                    commonService.saveOrUpdate(qres);
                    threadCounter.get().put(key, 0L);
                });
    }

    @Override
    public void logConsumer(MnsConsumerLogParamDto paramDto) throws Exception {
        String consumer = paramDto.getConsumerMnsQueue().getConsumer();
        String queueName = paramDto.getConsumerMnsQueue().getQueueName();
        String key = "C" + consumer + queueName;
        counterAdd(key);
        this.concurrentlog(key,
                () -> {
                    Long count = threadCounter.get().get(key);
                    MnsConsumLogCountEntity find = new MnsConsumLogCountEntity();
                    find.setConsumer(consumer);
                    find.setQueueName(queueName);
                    MnsConsumLogCountEntity qres = commonService.findOneEQ(MnsConsumLogCountEntity.class, find);
                    if (qres == null) {
                        qres = new MnsConsumLogCountEntity();
                        qres.setLhotelGroupCode(DefaultValue.LHOTEL_GROUP_CODE);
                        qres.setQueueName(queueName);
                        qres.setConsumer(consumer);
                        qres.setCount(0L);
                    }
                    long nCount = qres.getCount() + count;
                    qres.setCount(nCount);
                    commonService.saveOrUpdate(qres);
                    threadCounter.get().put(key, 0L);
                });
    }

    private void counterAdd(String key) {
        if (threadCounter.get() == null) {
            threadCounter.set(new HashMap<>());
        }
        if (null == threadCounter.get().get(key)) {
            synchronized (lockObj) {
                threadCounter.get().putIfAbsent(key, 0L);
            }
        }

        Long count = threadCounter.get().get(key);
        count++;
        threadCounter.get().put(key, count);
    }


    private interface Logger {
        void log() throws Exception;
    }
    private void concurrentlog(String key, Logger logger) throws Exception {
        Long count = threadCounter.get().get(key);
        Lock lock = mnsLogLock.obtain(key);
        if (count >= 100) { //大于100 记录一次
            if (lock.tryLock(30, TimeUnit.SECONDS)) {
                try {
                    logger.log();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
}
